package com.lm.activiti.boot.mq.rabbit;

import com.lm.activiti.boot.mq.common.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author lm_jcf
 * @date 2020/8/31 16:31
 */
public class Comsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_exchange_topic_queue";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        // 保证一次只分发一条消息
        channel.basicQos(1);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("read_1 msg：" + msg);

                channel.basicAck(envelope.getDeliveryTag(), false);

            }
        };

        /**
         * TODO autoAck = true 自动应答，一旦rabbitmq将消息分发给消费者，就会从内存删除该消息，当目标A消费者接收消息前处理错误，则会丢失该消息
         * TODO autoAck = false（默认） 手动应答，如若rabbitmq未收到消息回执，则会再次把消息发送给其他相同消费者
         * 当rabbitmq挂掉，消息一样会丢失，可做持久化操作
         *
         *
         */
        boolean autoAck = false; // 自动应答
        // 绑定队列
        channel.basicConsume(queueName, autoAck, consumer);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
